package org.apache.zookeeper.server.paxos2;

import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.zookeeper.proto.PrepareDataRequest;
import org.apache.zookeeper.server.DataTree;
import org.apache.zookeeper.server.PrepareDataCache;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

public class Storage {
    private static final Logger LOG = Logger.getLogger(Storage.class);

    private static Metadata meta = new Metadata();
    private BitSet updatedAfterFetch = new BitSet((int) (Configuration.totalSize / Configuration.blockSize));
    private HashMap<Integer, Shard> myStripes = new HashMap<Integer, Shard>();
    private FullExtraShard extra = new FullExtraShard();
    private DataCache cache = new DataCache(this);
    // private static SyncThread syncThread = new SyncThread();

    private static SocketManager sockManager = new SocketManager();
    private FetchThread fetchThread = new FetchThread();

    private DataTree dataTree;

    public Storage(List<Integer> stripes, DataTree dataTree) {
        this.dataTree = dataTree;
        for (Integer stripeId : stripes) {
            Shard s = new Shard(stripeId);
            myStripes.put(stripeId, s);
        }
        /*
           * syncThread.setStorage(this); if (!syncThread.isAlive())
           * syncThread.start();
           */
        sockManager.setStorage(this);
        if (!sockManager.isAlive())
            sockManager.start();
        fetchThread.start();
    }

    private boolean read(long blockId, byte[] data) {
        if (!meta.isValid(blockId))
            return false;
        else {
            readFromStripe(blockId, data);
            return true;
        }
        // return cache.read(blockId);
    }

    public synchronized boolean[] read(long[] blockId, byte[][] data) {
        if (!Configuration.isReadAvailable()) {
            System.out.println("read not available yet");
            boolean ret[] = new boolean[blockId.length];
            for (int i = 0; i < ret.length; i++)
                ret[i] = false;
            return ret;
        }
        long start = System.currentTimeMillis();
        TreeMap<Long, LinkedList<Integer>> sortBlockId = new TreeMap<Long, LinkedList<Integer>>();
        for (int i = 0; i < blockId.length; i++) {
            long physicalOffset = Mapping.keyToOffsetInShard(blockId[i]);
            if (!sortBlockId.containsKey(physicalOffset))
                sortBlockId.put(physicalOffset, new LinkedList<Integer>());
            sortBlockId.get(physicalOffset).addLast(i);
        }
        boolean ret[] = new boolean[blockId.length];
        int count = 0;
        for (LinkedList<Integer> tmp : sortBlockId.values()) {
            for (Integer index : tmp) {
                ret[index] = read(blockId[index], data[index]);
                count++;
                if (ret[index] == false)
                    System.out.println("invalid read " + blockId[index]);
            }
        }
        if (count != blockId.length)
            throw new RuntimeException("Missing some read");
        /*
           * boolean ret[] = new boolean[blockId.length]; for(int i=0;
           * i<blockId.length; i++){ ret[i] = read(blockId[i], data[i]); }
           */
        //System.out.println("read " + blockId.length + " data time=" + (System.currentTimeMillis() - start));
        return ret;
    }

    public synchronized boolean readWithMeta(long blockId, byte[] data,
                                             BlockMeta blockMeta) {
        if (!meta.isValid(blockId))
            return false;
        else {
            BlockMeta tmp = meta.getBlockMeta(blockId);
            if (tmp.zxid < blockMeta.zxid)
                return false;
            blockMeta.zxid = tmp.zxid;
            blockMeta.cxid = tmp.cxid;
            blockMeta.clientId = tmp.clientId;
            readFromStripe(blockId, data);
            return true;
        }
    }

    long totalTime = 0;
    int count = 0;
    long lastWriteTime = 0;
    long totalGap = 0;

    private void write(long blockId, byte[] data, BlockMeta blockMeta) {
        long tmp = System.currentTimeMillis();
        //if(this.totalBlocksToFetch > 0 && data==null)
        //    System.out.println("write "+blockId);
        if (count > 4)
            totalGap += (tmp - lastWriteTime);
        // System.out.println("gap "+(tmp-lastWriteTime));
        if (blockId == -1)
            return;
        this.updatedAfterFetch.set((int) blockId);
        if (data == null) {
            //System.out.println("set "+blockId+" as invalid");
            meta.setInvalid(blockId);
            meta.setBlockMeta(blockId, blockMeta);
            return;
        }
        writeToStripe(blockId, data);
        meta.setBlockMeta(blockId, blockMeta);
        meta.setValid(blockId);

        long time = System.currentTimeMillis() - tmp;
        if (count > 4)
            totalTime += time;
        count++;
        if (count % 1000 == 999) {
            System.out.println(" average = "
                    + ((double) totalTime / (double) count) + " gap="
                    + ((double) totalGap / (double) count));
            // count=0;
            // totalTime=0;
            // totalGap=0;
        }
        lastWriteTime = System.currentTimeMillis();
        // System.out.println("Write "+blockId+" "+data.length);
        // cache.write(blockId, data);
    }

    public synchronized void write(long[] blockId, byte[][] data,
                                   BlockMeta[] blockMetas) {
        /*
           * TreeMap<Long, LinkedList<Integer>> sortBlockId = new TreeMap<Long,
           * LinkedList<Integer>>(); for (int i = 0; i < blockId.length; i++) { if
           * (!sortBlockId.containsKey(blockId[i])) sortBlockId.put(blockId[i],
           * new LinkedList<Integer>()); sortBlockId.get(blockId[i]).addLast(i); }
           * int count = 0; for (LinkedList<Integer> tmp : sortBlockId.values()) {
           * for (Integer index : tmp) { write(blockId[index], data[index],
           * version[index]); count++; } } if (count != blockId.length) throw new
           * RuntimeException("Missing some read");
           */

        for (int i = 0; i < blockId.length; i++) {
            if (data[i] != null)
                Configuration.decreaseOutstanding();
            this.write(blockId[i], data[i], blockMetas[i]);
        }
    }

    public void close() {
        cache.close();
        for (Integer id : myStripes.keySet())
            myStripes.get(id).close();
        meta.close();

    }


    protected void readFromStripe(long blockId, byte[] data) {
        // return tmp;
        Shard s = myStripes.get(Mapping.keyToShard(blockId));
        if (s == null)
            extra.read(blockId, data);
        else
            s.read(blockId, data);
    }

    /*
      * public synchronized boolean isValid(long blockId) { if
      * (meta.getVersion(blockId) == -1) return false; else return true; }
      */

    public synchronized void setInvalid(long blockId) {
        if (Configuration.debug)
            System.out.println("SetInvalid " + blockId);
        meta.setInvalid(blockId);
        Shard s = myStripes.get(Mapping.keyToShard(blockId));
        if (s == null)
            extra.remove(blockId);
    }

    protected void writeToStripe(long blockId, byte[] data) {
        int stripeId = Mapping.keyToShard(blockId);
        if (!myStripes.containsKey(stripeId)) {
            // System.out.println("Block " + blockId + "stripeId " + stripeId
            // + "  should not belong to me " + myStripes);
            if (meta.isValid(blockId))
                extra.write(blockId, data, true);
            else
                extra.write(blockId, data, false);
            MemoryBuffer.deallocateBuffer(data);
            return;
        }
        Shard s = myStripes.get(Mapping.keyToShard(blockId));
        s.write(blockId, data);
        MemoryBuffer.deallocateBuffer(data);
    }

    public void serialize(OutputArchive oa, String tag) throws IOException {
        meta.sync();
        for (Integer id : myStripes.keySet())
            myStripes.get(id).serialize(oa, "shard");
        extra.serialize(oa, "shard");
    }

    public synchronized void deserialize(InputArchive ia, String tag)
            throws IOException {
        for (Integer id : myStripes.keySet())
            myStripes.get(id).deserialize(ia, "shard");
        extra.deserialize(ia, "shard");
    }

    public void getNewBlocks(long zxid, Storage storage, int sid) {
        Configuration.setBusy(true);
        System.out.println("processSync getNewBlocks " + zxid);
        meta.getNewBlocks(zxid, storage, sid);
        Configuration.setBusy(false);
    }

    private static class BlockInfo {
        public long block;
        public BlockMeta meta;
        public HashSet<Integer> validServers = new HashSet<Integer>();
        public HashSet<Integer> invalidServers = new HashSet<Integer>();

        public BlockInfo(long block, BlockMeta meta) {
            this.block = block;
            this.meta = meta;
        }
    }

    private synchronized void fetchMetadata(long zxid) {
        System.out.println("recovery fetch metadata this=" + zxid);
        boolean succeed = false;
        int sid = 1;

        long targetZxid = 0;
        while (!succeed) {
            sid = (sid + 1) % Configuration.servers.length;
            if (sid + 1 == Configuration.serverId)
                continue;
            try {
                System.out.println("Try to sync with "
                        + Configuration.servers[sid]);
                StartSync startSync = new StartSync();
                startSync.zxid = zxid;
                if (!sockManager.send(sid, startSync))
                    continue;
                boolean end = false;
                while (!end) {
                    while (startSyncResponse.size() == 0)
                        this.wait();
                    StartSyncResponse resp = startSyncResponse.removeFirst();
                    if(resp.zxid>targetZxid)
                        targetZxid = resp.zxid;

                    System.out.println("Got StartSyncResponse from "
                            + Configuration.servers[sid] + " length="
                            + resp.blocks.length + " zxid=" + resp.zxid+" last="+resp.last);

                    for (int j = 0; j < resp.blocks.length; j++) {
                        BlockMeta meta = new BlockMeta(ByteBuffer
                                .wrap(resp.blockMetas[j]));
                        PrepareDataRequest p = PrepareDataCache.get(meta.clientId,
                                meta.cxid);
                        if (p != null) {
                            this.write(resp.blocks[j], p.getData(), meta);
                        } else {
                            this.write(resp.blocks[j], null, meta);
                        }
                    }
                    if(resp.last)
                        end = true;
                }
                dataTree.lastProcessedZxid = targetZxid;
                System.out.println("lastProcessedZxid="+targetZxid);
                succeed = true;
            } catch (Exception e) {
                LOG.warn("catch up", e);
                // e.printStackTrace();
                continue;
            }
        }

    }

    private static ArrayList<Integer> determineTargetServers(int shardId) {
        List<Integer> tmp = Mapping.shardToNode(shardId, null);
        tmp.remove(new Integer(Configuration.serverId));
        return new ArrayList<Integer>(tmp);
    }

    private synchronized void recover(long zxid, boolean blocking) {
        if (this.totalBlocksToFetch > 0) {
            System.out.println("Already in recovery");
            return;
        }
        this.fetchMetadata(zxid);
        LinkedList<FetchBlocks> toFetch = new LinkedList<FetchBlocks>();
        long total = 0;
        if (Configuration.rereplicate)
            total = rebuild(toFetch);
        else
            total = scanInvalidBlocks(toFetch);
        System.out.println("recover needs to fetch " + total + " blocks");
        if (total == 0)
            return;
        this.startFetching((int) total);
        for (FetchBlocks fetchBlock : toFetch) {
            System.out.println("send " + fetchBlock + " to " + fetchBlock.targetSid);
            sockManager.send(fetchBlock.targetSid - 1, fetchBlock);
        }
        if (blocking)
            this.waitForFetching();
        Configuration.rereplicate = false;
        System.out.println("recover finishes " + zxid);
    }

    private synchronized void fetchInvalid() {
        if (this.totalBlocksToFetch > 0) {
            System.out.println("Already in recovery");
            return;
        }
        LinkedList<FetchBlocks> toFetch = new LinkedList<FetchBlocks>();
        long total = 0;
        total = scanInvalidBlocks(toFetch);
        System.out.println("fetchInvalid needs to fetch " + total + " blocks");
        if (total == 0)
            return;
        this.startFetching((int) total);
        for (FetchBlocks fetchBlock : toFetch) {
            System.out.println("send " + fetchBlock + " to " + fetchBlock.targetSid);
            sockManager.send(fetchBlock.targetSid - 1, fetchBlock);
        }

    }

    private synchronized long scanInvalidBlocks(LinkedList<FetchBlocks> ret) {
        long targetZxid = dataTree.lastProcessedZxid;
        long total = 0;
        //Find incomplete blocks
        if (Configuration.writeQuorumSize < Configuration.noOfStripes) {
            HashMap<Integer, LinkedList<ArrayList<Long>>> blocksToFetch = new HashMap<Integer, LinkedList<ArrayList<Long>>>();

            for (Integer shardId : myStripes.keySet()) {
                blocksToFetch.put(shardId, new LinkedList<ArrayList<Long>>());
                blocksToFetch.get(shardId).addLast(new ArrayList<Long>(4000));
            }
            for (long i = 0; i < Configuration.totalSize / Configuration.blockSize; i++) {
                int shardId = Mapping.keyToShard(i);
                if (!meta.isValid(i) && myStripes.containsKey(shardId)) {
                    total++;
                    ArrayList<Long> tmp = blocksToFetch.get(shardId).getLast();
                    tmp.add(i);
                    if (tmp.size() >= 4000) {
                        blocksToFetch.get(shardId).addLast(new ArrayList<Long>(4000));
                    }
                }
            }

            for (Integer shardId : blocksToFetch.keySet()) {
                ArrayList<Integer> targetServers = determineTargetServers(shardId);
                int size = targetServers.size();
                Random r = new Random(System.currentTimeMillis());
                LinkedList<ArrayList<Long>> targets = blocksToFetch.get(shardId);
                for (ArrayList<Long> target : targets) {
                    if (target.size() == 0)
                        continue;
                    long[] tmp = new long[target.size()];
                    for (int i = 0; i < tmp.length; i++)
                        tmp[i] = target.get(i);
                    FetchBlocks fetchBlocks = new FetchBlocks();
                    fetchBlocks.sid = Configuration.serverId - 1;
                    fetchBlocks.blockIds = tmp;
                    fetchBlocks.zxid = targetZxid;
                    fetchBlocks.targetSid = targetServers.get(r.nextInt(size));
                    ret.addLast(fetchBlocks);
                }
            }
        } else {
            ArrayList<Long> target = new ArrayList<Long>(4000);
            int replica = 0;
            for (long i = 0; i < Configuration.totalSize / Configuration.blockSize; i++) {
                if (!meta.isValid(i)) {
                    total++;
                    target.add(i);
                    if (target.size() >= 4000) {
                        long[] tmp = new long[target.size()];
                        for (int j = 0; j < tmp.length; j++)
                            tmp[j] = target.get(j);
                        FetchBlocks fetchBlocks = new FetchBlocks();
                        fetchBlocks.sid = Configuration.serverId - 1;
                        fetchBlocks.blockIds = tmp;
                        fetchBlocks.zxid = targetZxid;
                        replica = (replica + 1) % Configuration.servers.length;
                        while (replica == Configuration.serverId - 1)
                            replica = (replica + 1) % Configuration.servers.length;
                        fetchBlocks.targetSid = replica + 1;
                        ret.addLast(fetchBlocks);
                        target.clear();
                    }

                }
            }

            if (target.size() != 0) {
                long[] tmp = new long[target.size()];
                for (int j = 0; j < tmp.length; j++)
                    tmp[j] = target.get(j);
                FetchBlocks fetchBlocks = new FetchBlocks();
                fetchBlocks.sid = Configuration.serverId - 1;
                fetchBlocks.blockIds = tmp;
                fetchBlocks.zxid = targetZxid;
                replica = (replica + 1) % Configuration.servers.length;
                while (replica == Configuration.serverId - 1)
                    replica = (replica + 1) % Configuration.servers.length;
                fetchBlocks.targetSid = replica + 1;
                ret.addLast(fetchBlocks);
            }
        }
        return total;
    }

    private synchronized long rebuild(LinkedList<FetchBlocks> ret) {
        long targetZxid = dataTree.lastProcessedZxid;
        long total = 0;
        if (Configuration.writeQuorumSize < Configuration.noOfStripes) {
            List<Integer> shards = Mapping.listShards(Configuration.serverId);
            total = Configuration.totalSize
                    / Configuration.blockSize / Configuration.noOfStripes
                    * shards.size();
            for (Integer shardId : shards) {
                System.out.println("need to reReplicate shard " + shardId);
                FetchBlocks fetchBlocks = new FetchBlocks();
                fetchBlocks.startBlock = Mapping.getBlock(shardId, 0);
                fetchBlocks.endBlock = Mapping.getBlock(shardId,
                        Configuration.totalSize / Configuration.blockSize
                                / Configuration.noOfStripes - 1);
                fetchBlocks.interval = Configuration.noOfStripes;
                fetchBlocks.sid = Configuration.serverId - 1;
                fetchBlocks.zxid = targetZxid;
                int replica = (shardId - 1 + Configuration.servers.length)
                        % Configuration.servers.length;
                while (replica == Configuration.serverId - 1) {
                    replica = (replica - 1 + Configuration.servers.length)
                            % Configuration.servers.length;
                }
                fetchBlocks.targetSid = replica + 1;
                ret.addLast(fetchBlocks);
            }
        } else {
            total = Configuration.totalSize / Configuration.blockSize;
            long numBlocksEachReplica = total / (Configuration.servers.length - 1);
            int index = 0;
            for (int i = 0; i < Configuration.servers.length; i++) {
                if (i == Configuration.serverId - 1)
                    continue;
                FetchBlocks fetchBlocks = new FetchBlocks();
                fetchBlocks.startBlock = index * numBlocksEachReplica;
                fetchBlocks.endBlock = (index + 1) * numBlocksEachReplica - 1;
                fetchBlocks.interval = 1;
                fetchBlocks.sid = Configuration.serverId - 1;
                fetchBlocks.zxid = targetZxid;
                if (!Configuration.simulateCheapPaxos) {
                    fetchBlocks.targetSid = i + 1;
                } else {
                    fetchBlocks.targetSid = 2 + 1;
                }
                ret.addLast(fetchBlocks);
                index++;
            }
        }
        return total;
    }


    private int totalBlocksToFetch = 0;

    private synchronized void startFetching(int num) {
        Configuration.setInRecovery(true);
        System.out.println("start fetching " + num + " blocks");
        if (totalBlocksToFetch != 0)
            throw new RuntimeException("Cannot start fetch while already fetching");
        this.updatedAfterFetch.clear();
        totalBlocksToFetch = num;
    }

    private synchronized void updateFetching(FetchBlocksResponse resp) {
        totalBlocksToFetch -= resp.blockIds.length;
        System.out.println("remaining=" + totalBlocksToFetch);
        if (totalBlocksToFetch == 0) {
            System.out.println("fetch completes at " + System.currentTimeMillis());
            Configuration.setInRecovery(false);
            this.notifyAll();
            this.fetchInvalid();
        }
    }

    private synchronized void waitForFetching() {
        while (totalBlocksToFetch > 0) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                LOG.warn("interrupted", e);
            }
        }
    }


    public void syncWithOthers(long zxid) {
        System.out.println("Start syncWithOthers " + zxid);
        if (Configuration.writeQuorumSize == Configuration.noOfStripes)
            recover(zxid, true);
        else
            recover(zxid, false);
        Configuration.outstandingCount = 0;
    }

    public void handleMessage(Object msg, int sid) {
        if (msg instanceof StartSync) {
            handlStartSync((StartSync) msg, sid);
        } else if (msg instanceof StartSyncResponse) {
            handleStartSyncResponse((StartSyncResponse) msg, sid);
        } else if (msg instanceof FetchBlocks) {
            handleFetchBlocks((FetchBlocks) msg, sid);
        } else if (msg instanceof FetchBlocksResponse) {
            handleFetchBlocksResponse((FetchBlocksResponse) msg, sid);
        }
    }

    protected void sendStartSyncReply(LinkedList<Entry<Long, byte[]>> ret, boolean last, int sid) {
        StartSyncResponse resp = new StartSyncResponse();
        resp.blocks = new long[ret.size()];
        resp.blockMetas = new byte[ret.size()][];
        long lastZxid = -1;
        int index = 0;
        while (!ret.isEmpty()) {
            Entry<Long, byte[]> e = ret.removeFirst();
            resp.blocks[index] = e.getKey();
            resp.blockMetas[index] = e.getValue();
            BlockMeta tmp = new BlockMeta(ByteBuffer.wrap(resp.blockMetas[index]));
            if (tmp.zxid > lastZxid)
                lastZxid = tmp.zxid;
            index++;
        }

        resp.last = last;
        resp.zxid = lastZxid;
        resp.sid = Configuration.serverId - 1;
        System.out.println("processSync start sending StartSyncResponse size" + resp.blocks.length + " last=" + lastZxid);

        sockManager.send(sid, resp);
    }

    private void handlStartSync(StartSync startSync, int sid) {
	System.out.println("processSync start StartSyncResponse "+startSync.zxid+" "+dataTree.lastProcessedZxid);
        this.getNewBlocks(startSync.zxid, this, sid);
        System.out.println("processSync end sending StartSyncResponse");
    }

    private LinkedList<StartSyncResponse> startSyncResponse = new LinkedList<StartSyncResponse>();

    private synchronized void handleStartSyncResponse(StartSyncResponse startSyncResponse,
                                                      int sid) {
        this.startSyncResponse.addLast(startSyncResponse);
        this.notifyAll();
    }

    private void handleFetchBlocks(FetchBlocks fetchBlocks, int sid) {
        if (fetchBlocks.zxid > dataTree.lastProcessedZxid) {
            System.out.println("receive a future fetch " + fetchBlocks.zxid + " current=" + dataTree.lastProcessedZxid);
            //return;
        }
        fetchThread.fetch(fetchBlocks);
    }

    private synchronized void fetchBlocks(FetchBlocksResponse resp, int target) {
        resp.sid = Configuration.serverId - 1;
        System.out.println("fetch " + resp.blockIds.length
                + " blocks first is " + resp.blockIds[0]);
        for (int i = 0; i < resp.blockIds.length; i++) {
            long blockId = resp.blockIds[i];
            byte[] data = new byte[Configuration.blockSize];
            boolean valid = read(blockId, data);
            if (valid)
                resp.data[i] = data;
            else {
                //System.out.println(blockId+" is invalid");
                resp.data[i] = null;
            }
        }
        sockManager.send(target, resp);

    }

    private LinkedBlockingQueue<FetchBlocksResponse> fetchBlocksResponseQueue = new LinkedBlockingQueue<FetchBlocksResponse>(
            20);

    private synchronized void handleFetchBlocksResponse(FetchBlocksResponse resp, int sid) {
        System.out.println("Got a FetchBlocksResponse length="
                + resp.blockIds.length + " first=" + resp.blockIds[0]);
        long start = System.currentTimeMillis();
        for (int i = 0; i < resp.blockIds.length; i++) {
            if (this.updatedAfterFetch.get((int) resp.blockIds[i]))
                continue;
            if (resp.data[i] != null) {

                this.meta.setValid(resp.blockIds[i]);
                writeToStripe(resp.blockIds[i], resp.data[i]);
            } else {
                this.meta.setInvalid(resp.blockIds[i]);
                //throw new RuntimeException(
                //		"Got an incomplete reply!!!!! Not implemented yet");
            }
        }
        System.out.println("fetch time=" + (System.currentTimeMillis() - start));
        this.updateFetching(resp);
        /*
           * this.numBlocksToFetch -= fetchBlocksResponse.blockIds.length; if
           * (this.numBlocksToFetch == 0) { Configuration.setReadAvailable(true);
           * System.out.println("Re-replicate ends at " +
           * System.currentTimeMillis()); }
           */
    }

    private class FetchThread extends Thread {
        private LinkedBlockingQueue<FetchBlocks> reqs = new LinkedBlockingQueue<FetchBlocks>();

        public void fetch(FetchBlocks req) {
            try {
                reqs.put(req);
            } catch (InterruptedException e) {
                LOG.warn("interrupted", e);
            }
        }

        public void run() {
            while (true) {
                FetchBlocks req = null;
                try {
                    req = reqs.take();
                    while (req.zxid > dataTree.lastProcessedZxid) {
                        Thread.sleep(100);
                    }
                    if (Configuration.simulateCheapPaxos) {
                    } else if (Configuration.writeQuorumSize < Configuration.noOfStripes)
                        Thread.sleep(600);
                    else
                        Thread.sleep(600);
                } catch (InterruptedException e) {
                    LOG.warn("interrupted", e);
                    return;
                }

                // This is a fetch shard command
                if (req.startBlock != -1) {
                    FetchBlocksResponse resp = null;
                    int remaining = (int) ((req.endBlock - req.startBlock) / req.interval + 1);
                    long now = req.startBlock;
                    while (remaining > 0) {
                        if (remaining > 4000)
                            resp = new FetchBlocksResponse(4000);
                        else
                            resp = new FetchBlocksResponse(remaining);
                        resp.sid = Configuration.serverId - 1;

                        for (int i = 0; i < resp.blockIds.length; i++) {
                            resp.blockIds[i] = now;
                            remaining--;
                            now += req.interval;
                        }
                        fetchBlocks(resp, req.sid);
                        try {
                            if (Configuration.simulateCheapPaxos) {
                            } else if (Configuration.writeQuorumSize < Configuration.noOfStripes)
                                Thread.sleep(600);
                            else
                                Thread.sleep(600);
                        } catch (InterruptedException e) {
                        }
                    }
                }
                // This is a fetch some blocks
                else {
                    FetchBlocksResponse resp = new FetchBlocksResponse(
                            req.blockIds.length);
                    resp.sid = Configuration.serverId - 1;
                    resp.blockIds = req.blockIds;
                    fetchBlocks(resp, req.sid);
                }
            }

        }
    }
}
